package com.raylu.realtime.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.io.IOException;
import java.util.Properties;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2021/12/24.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
public class KafkaSourceUtil {
        private static Properties properties;

    static {
        try {
            properties = PropertiesUtil.load("config.properties");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static FlinkKafkaConsumer<String> getKafkaSource(String topic,String groupId)  {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getProperty("base.log.app.kafka.bootstrap-server"));
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);
        return new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(),props);
    }

    public static String getConnectorDDl(String topic,String groupId,String formatStr){
        String ddl = "  'connector' = 'kafka',\n" +
                     "  'topic' = '"+topic+"',\n" +
                     "  'properties.bootstrap.servers' = '"+properties.getProperty("base.log.app.kafka.bootstrap-server")+"',\n" +
                     "  'properties.group.id' = '"+groupId+"',\n" +
                     "  'scan.startup.mode' = 'latest-offset',\n" + //group-offsets/earliest-offset/latest-offset
                     "  'format' = '"+formatStr+"' \n";
        return ddl;
    }


}
